package com.lagou;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 用于统计接口调用的耗时，并计算输出TP90和TP99指标
 */
@Activate(group = {CommonConstants.CONSUMER})
public class DubboApiFilter implements Filter {
    /**
     * 接口调用记录
     */
    private Map<String, InvocationInfo> invocationInfoMap = new ConcurrentHashMap<>();

    /**
     * 线程池
     */
    private ScheduledExecutorService threadPool = Executors.newSingleThreadScheduledExecutor();

    /**
     * 构造DubboApiFilter对象
     */
    public DubboApiFilter() {
        threadPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    reportStatInfo();
                    System.out.println("end");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }, 5, 5, TimeUnit.SECONDS);
    }

    /**
     * 打印接口调用的性能统计结果
     */
    private void reportStatInfo() {
        System.out.println("Report start");
        Set<Map.Entry<String, InvocationInfo>> entrySet = invocationInfoMap.entrySet();
        for (Map.Entry<String, InvocationInfo> entry : entrySet) {
            String methodName = entry.getKey();
            InvocationInfo invocationInfo = entry.getValue();

            List<Long> responseTimes = invocationInfo.getResponseTimes();
            if (responseTimes.isEmpty()) {
                continue;
            }

            List<Long> sortedReponseTimes = new LinkedList<>(responseTimes);
            Collections.sort(sortedReponseTimes);

            int size = sortedReponseTimes.size();
            int tp90Idx = (int)(size * 0.9 - 1);
            int tp99Idx = (int)(size * 0.99 - 1);

            System.out.println(String.format("Method: %s TP90: %d TP99: %d", methodName,
                    sortedReponseTimes.get(tp90Idx), sortedReponseTimes.get(tp99Idx)));
        }
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        Result result = null;

        try {
            result = invoker.invoke(invocation);
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        long endTime = System.currentTimeMillis();
        recordInvocationInfo(invocation.getMethodName(), startTime, endTime);

        return result;
    }

    /**
     * 记录接口调用的统计信息
     *
     * @param methodName 接口名
     * @param startTime 调用开始时间
     * @param endTime 调用结束时间
     */
    private void recordInvocationInfo(String methodName, long startTime, long endTime) {
        InvocationInfo invocationInfo = invocationInfoMap.get(methodName);
        if (invocationInfo == null) {
            invocationInfoMap.putIfAbsent(methodName, new InvocationInfo());
            invocationInfo = invocationInfoMap.get(methodName);
            invocationInfo.setStartTime(startTime);
        }

        List<Long> responseTimes = invocationInfo.getResponseTimes();
        if (startTime > (invocationInfo.getStartTime() + 60 * 1000)) {
            responseTimes.clear();
            invocationInfo.setStartTime(startTime);
        }

        responseTimes.add(endTime - startTime);
    }
}
